/*
 * @Author: 来自火星的码农 15122322+heyzhi@user.noreply.gitee.com
 * @Date: 2025-07-24 16:35:27
 * @LastEditors: 来自火星的码农 15122322+heyzhi@user.noreply.gitee.com
 * @LastEditTime: 2025-09-19 21:35:40
 * @FilePath: /server/workspace/src/base/iomanager.cc
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
#include"iomanager.h"
#include"util.h"
#include"marco.h"
#include <cerrno>
#include <cstdint>
#include <ctime>
#include <functional>
#include <netinet/in.h>
#include<sys/epoll.h>
#include<string.h>
#include<fcntl.h>
#include <vector>
#include"Log.h"
#include"scheduler.h"
#include "timer.h"
namespace Yizhi {
    Logger::ptr logger=SERVER_LOG_NAME("system");


    /// @brief 重置事件上下文
    void IOManager::FdContext::Eventcontext::reset(){
        cb=nullptr;
        m_scheduler=nullptr;
        fiber.reset();
    }


    ///根据事件类型返回对应的事件上下文
    IOManager::FdContext::Eventcontext& IOManager::FdContext::GetEvent(Event event){
        switch (event)
        {
        case READ:
            return Read;
            break;
        case WRITE:
            return Write;
            break;
        default:
            break;
        }
        throw std::invalid_argument("getContext invalid event");
    }

    void IOManager::FdContext::resetEventcontext(Event event){
        switch (event)
        {
        case READ:
            /* code */
            Read.reset();
            return;
        case WRITE:
            Write.reset();
            return;
        default:
            break;
        }
    }
    void IOManager::FdContext::triggerEvent(Event event){
        if((event & m_events)==0){

            return;
        }
        m_events=(Event)(m_events & (~event));
        Eventcontext& ectx=GetEvent(event);

        if(ectx.cb){
           ectx.m_scheduler->scheduler(ectx.cb);
        }
        else{
           ectx.m_scheduler->scheduler(ectx.fiber);
        }

        resetEventcontext(event);
    }
    IOManager::IOManager(int threads,bool user_caller,std::string name)
    :Scheduler(threads,name,user_caller){
        //创建epoll文件句柄
        m_epfd=epoll_create(500);
        
        SERVER_ASSERT(m_epfd>0);

        //创建管道
        int rt=pipe(m_tickleFds);

        SERVER_ASSERT(!rt);
        

        //breif向epoll注册pipe的可读事件，当pipe读事件触发时，唤醒tickle函数中的epoll_out
        epoll_event event;

        //初始化event,确保不会包含未定义的数据值
        memset(&event,0,sizeof(epoll_event));

        //通过使用按位或运算符|,可以同时监听多种事件，event.events将同时监听管道读端的可读和可写事件。
        event.events=EPOLLIN|EPOLLOUT;

        event.data.fd=m_tickleFds[0];

        rt=fcntl(m_tickleFds[0],F_SETFL,O_NONBLOCK);
        SERVER_ASSERT(!rt);

        //注册监听pipe读事件
        rt=epoll_ctl(m_epfd,EPOLL_CTL_ADD,m_tickleFds[0],&event);
        SERVER_ASSERT(!rt);

        //初始化m_fdcontext
        resizeFdcontexts(32);
        
        //开始
        start();
    }


    void IOManager::resizeFdcontexts(size_t size){
       // RWMutexType::ReadLock rlock(m_rwmutex);
        if(size<=m_FdContexts.size()){
            return;
        }
        else{
            m_FdContexts.resize(size);
            for(size_t i=0;i<m_FdContexts.size();i++){
                if(!m_FdContexts[i]){
                    m_FdContexts[i]=new FdContext();
                    m_FdContexts[i]->fd=i;
                }
            }
        }
    }

    void IOManager::tickle(){
        SERVER_LOG_INFO(logger)<<"tickle";
        if(!hasIdleThreads()){
            return;
        }
        int rt=write(m_tickleFds[1],"T",1);

        SERVER_ASSERT(rt==1);
    }

    void IOManager::onTimerInsertFront(){
        tickle();
    }



    void IOManager::idle(){
        //声明epoll_event集合，一次epoll最大处理256个事件
        SERVER_LOG_DEBUG(logger)<<"idle";
        const uint64_t MAX_EVENTS=256;
        
        epoll_event* events=new epoll_event[MAX_EVENTS]();
        std::shared_ptr<epoll_event> shared_events(events,[](epoll_event* ptr){
            delete[] ptr;
        });
        
        while (true){
            uint64_t nexttime=0;
            if(stopping(nexttime)){
                break;
                SERVER_LOG_DEBUG(logger)<<"idle has stoping exit";
            }

            // static int MAX_TIMEOUT=5000;
            // int rt=epoll_wait(m_epfd,events,MAX_EVENTS,MAX_TIMEOUT);

            // if(rt<0){
            //     if(errno==EINTR){
            //         continue;
            //     }
            //     break;
            // }
         
            int rt=0;
            do{
                static int MAX_TIMEOUT=5000;
                if(nexttime!=~0ull){
                    nexttime=(int)nexttime>MAX_TIMEOUT?MAX_TIMEOUT:nexttime;
                }
                else{
                    nexttime=MAX_TIMEOUT;
                }
                rt=epoll_wait(m_epfd,events,MAX_EVENTS,(int)nexttime);

                if(rt<0&&errno==EINTR){

                }
                else{
                    break;
                }
            }while(true);

            std::vector<std::function<void()>> cbs;
            listExpiredCb(cbs);
            if(!cbs.empty()){
               Scheduler::scheduler(cbs.begin(),cbs.end());
               
                cbs.clear();
            }

            for(int i=0;i<rt;i++){
                epoll_event &event=events[i];
                if(event.data.fd==m_tickleFds[0]){
                    uint8_t dummy[256];
                    while (read(m_tickleFds[1],dummy,sizeof(dummy))>0);
                    continue;
                }

                FdContext* fd_ctx=(FdContext*)event.data.ptr;
                Event real_events=NONE;
                FdContext::MutexType::Lock(fd_ctx->m_mutex);

                if(event.events&(EPOLLERR|EPOLLHUP)){
                    event.events|=(EPOLLIN|EPOLLOUT)&fd_ctx->m_events;
                }

                real_events=(Event)(((READ|WRITE)&fd_ctx->m_events)&event.events);
                
                if((event.events & real_events)==NONE){
                    continue;
                }

                
                int left_events=(fd_ctx->m_events&~real_events);
                int op=left_events?EPOLL_CTL_MOD:EPOLL_CTL_DEL;

                event.events=EPOLLET|left_events;

                int rt2=epoll_ctl(m_epfd,op,fd_ctx->fd,events);

                if(rt2){
                    continue;
                }

                if(real_events&READ){
                    fd_ctx->triggerEvent(real_events);
                    m_pendingEventCount--;
                }
                if(real_events&WRITE){
                    fd_ctx->triggerEvent(real_events);
                    m_pendingEventCount--;
                }
            }
            Fiber::YieldToHold();
        }
      

    }

    bool IOManager::isStoped(){
        uint64_t timeout=0;
        return stopping(timeout);
    }


    bool IOManager::stopping(uint64_t timeout){
        timeout=Yizhi::TimerManager::GetNextTimer();

        return timeout==~0ull
        &&Yizhi::Scheduler::isStoped()
        &&m_pendingEventCount==0;
    }


    int IOManager::addEvent(int fd,Event event,std::function<void()>cb){
        FdContext* fd_ctx=nullptr;
        RWMutexType::ReadLock rlock(m_rwmutex);
        /// @brief 判断当前fd是否存在于数组中，如果不存在，则扩容
        if(fd<m_FdContexts.size()){
            fd_ctx=m_FdContexts[fd];
            rlock.unlock();
        }
        else{
            rlock.unlock();
            RWMutexType::WriteLock wlock(m_rwmutex);
            resizeFdcontexts(fd*1.5);
            fd_ctx=m_FdContexts[fd];
        }

        MutexType::Lock lock(fd_ctx->m_mutex);

        //当传入的事件重复
        if(fd_ctx->m_events & event){
            SERVER_ASSERT(false);
        }
        

        ///当仍有事件需要处理
        int op=fd_ctx->m_events?EPOLL_CTL_MOD:EPOLL_CTL_ADD;
        

        epoll_event epevent={0};

        epevent.events=fd_ctx->m_events|event|EPOLLET;
        epevent.data.ptr=fd_ctx;
        int rt=epoll_ctl(m_epfd,op,fd,&epevent);

        //失败
        if(rt){
            return -1;
        }

        //更新fd所执行的操作
        fd_ctx->m_events=(Event)(fd_ctx->m_events|event);
        m_pendingEventCount++;

        
        FdContext::Eventcontext& ectx=fd_ctx->GetEvent(event);
        if(cb){
            ectx.cb.swap(cb);
        }
        else{
            ectx.fiber=Fiber::GetThis();
        }
        
        ectx.m_scheduler=Scheduler::GetThis()?Scheduler::GetThis():nullptr;
        return 0;
    }

    bool IOManager::delEvent(int fd,Event event){
        //读锁保护
        RWMutexType::ReadLock rlock(m_rwmutex);
        //fd不存在
        if(fd>(int)m_FdContexts.size()){
            return false;
        }
        FdContext* fdctx=m_FdContexts[fd];

        rlock.unlock();
        MutexType::Lock lock(fdctx->m_mutex);

        //需要修改的事件不存在
        if(!(event & fdctx->m_events)){
            return false; 
        }

        Event new_event=(Event)(fdctx->m_events &~ event);
        int op=new_event?EPOLL_CTL_MOD:EPOLL_CTL_DEL;
        epoll_event epevent;
        epevent.events=EPOLLET|new_event;
        epevent.data.ptr=fdctx;
        int rt=epoll_ctl(m_epfd,op,fdctx->fd,&epevent);

        if(rt){

            return false;
        }
        --m_pendingEventCount;
        fdctx->m_events=new_event;
        fdctx->resetEventcontext(new_event);
        return true;
        
    }

    bool IOManager::cancleEvent(int fd,Event event){
        //读锁保护
        RWMutexType::ReadLock rlock(m_rwmutex);
        //fd不存在
        if(fd>(int)m_FdContexts.size()){
            return false;
        }
        FdContext* fdctx=m_FdContexts[fd];

        rlock.unlock();
        MutexType::Lock lock(fdctx->m_mutex);

        //需要修改的事件不存在
        if(!(event & fdctx->m_events)){
            return false; 
        }

        Event new_event=(Event)(fdctx->m_events &~ event);
        int op=new_event?EPOLL_CTL_MOD:EPOLL_CTL_DEL;
        epoll_event epevent;
        epevent.events=EPOLLET|new_event;
        epevent.data.ptr=fdctx;
        int rt=epoll_ctl(m_epfd,op,fdctx->fd,&epevent);

        if(rt){

            return false;
        }
        --m_pendingEventCount;
        fdctx->triggerEvent(event);
        return true;
    }

    bool IOManager::cancleAll(int fd){
        //读锁保护
        RWMutexType::ReadLock rlock(m_rwmutex);
        //fd不存在
        if(fd>(int)m_FdContexts.size()){
            return false;
        }
        FdContext* fdctx=m_FdContexts[fd];

        rlock.unlock();
        MutexType::Lock lock(fdctx->m_mutex);

        //需要修改的事件不存在
       

       
        int op=EPOLL_CTL_DEL;
        epoll_event epevent;
        epevent.events={0};
        epevent.data.ptr=fdctx;
        int rt=epoll_ctl(m_epfd,op,fdctx->fd,&epevent);

        if(rt){

            return false;
        }
        


        if(fdctx->m_events & READ){
            fdctx->triggerEvent(READ);
            m_pendingEventCount--;
        }

        if(fdctx->m_events & WRITE){
            fdctx->triggerEvent(WRITE);
            m_pendingEventCount--;
        }
        
        SERVER_ASSERT(fdctx->m_events==0);
        return true;
    }



    IOManager::~IOManager(){
        stop();
        close(m_epfd);
        close(m_tickleFds[0]);
        close(m_tickleFds[1]);

        for(size_t i=0;i<m_FdContexts.size();i++){
            if(m_FdContexts[i]){
                delete m_FdContexts[i];
            }
           
        }
    }

    IOManager* IOManager::GetThis(){
        return dynamic_cast<IOManager*>(Scheduler::GetThis());
    }

}